package com.nhncorp.pleiades;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import junit.framework.TestCase;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jgroups.stack.GossipRouter;

import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.client.LockServiceClientFactory;
import com.nhncorp.pleiades.client.PleiadesUser;
import com.nhncorp.pleiades.conf.PleiadesConfiguration;
import com.nhncorp.pleiades.ens.EventManager;
import com.nhncorp.pleiades.master.ClusterManager;
import com.nhncorp.pleiades.protocol.LockMetaData;
import com.nhncorp.pleiades.util.Sleeper;

public class PleiadesTestCase extends TestCase {
  
  public static final Log LOG = LogFactory.getLog(PleiadesTestCase.class
      .getName()); 
  protected PleiadesConfiguration conf;
  protected LockService pleiadesClient1;
  protected LockService pleiadesClient2;
  protected LockService pleiadesClient3;
  protected ClusterManager clusterManager;
  protected Thread clusterThread;
  protected Map<String, ClusterManager> clusterManagers;
  protected Map<String, PleiadesConfiguration> clusterManagerConfs;
  protected Map<String, Thread> clusterManagerThreads;
  protected EventManager eventManager;
  protected Thread eventThread;
  protected Map<String, EventManager> eventManagers;
  protected Map<String, Thread> eventManagerThreads;
  private int numberOfServers = 0;
  AtomicBoolean stopRequested;
  protected Sleeper sleeper;
  private GossipRouter firstRouter;
  private GossipRouter secondRouter;
  static final int portOfRouter1 = 12001;
  static final int portOfRouter2 = 12002;
  static final String bind_addr = "clustermanager.nhncorp.com";
  
  public PleiadesTestCase() {
    this.clusterManagers = new HashMap<String, ClusterManager>();
    this.clusterManagerConfs = new HashMap<String, PleiadesConfiguration>();
    this.clusterManagerThreads = new HashMap<String, Thread>();
    this.eventManagers = new HashMap<String, EventManager>();
    this.eventManagerThreads = new HashMap<String, Thread>();
    stopRequested = new AtomicBoolean(false);
    sleeper = new Sleeper(1000, stopRequested);
  }
  
  public void startCluster() {
    try {
      
      conf = new PleiadesConfiguration();
      if( conf.getBoolean("pleiades.mode.splitBrain", false) ) {
        startFirstRouter();
      }
      
      if( !conf.getBoolean("pleiades.server.outside", false) ) {
        this.numberOfServers = conf.getInt("pleiades.numberOfServers", 3);
        startClusterManagers(numberOfServers);
        startEventManagers(numberOfServers);
      }
      
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
  
  public void stopCluster() {
    try {
      if( !conf.getBoolean("pleiades.server.outside", false) ) {
        stopClusterManagers();
        stopEventManagers();        
      }
      if (conf.getBoolean("pleiades.mode.splitBrain", false)) {
        stopFirstRouter();
      }
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
  
  public void stopClusterMember(int serverNum) {
    int number = 0;
    for (Map.Entry<String, ClusterManager> entry : clusterManagers.entrySet()) {
      if(number == serverNum) {
        entry.getValue().shutdown();
      }
      ++number;
    }
    number = 0;
    for (Map.Entry<String, Thread> entry : clusterManagerThreads.entrySet()) {
      if(number == serverNum) {
        entry.getValue().interrupt();
      }
      ++number;
    }
  }
  
  public void startClient() {
    try {
      pleiadesClient1 = LockServiceClientFactory.newInstance(new PleiadesUser("default", "host#1"), conf);
      pleiadesClient2 = LockServiceClientFactory.newInstance(new PleiadesUser("default", "host#2"), conf);
      pleiadesClient3 = LockServiceClientFactory.newInstance(new PleiadesUser("default", "host#3"), conf);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
  
  public void stopClient() {
    pleiadesClient1.close();
    pleiadesClient2.close();
    pleiadesClient3.close();
  }
  
  protected String[] startClusterManagers(int numOfServers) throws IOException {
    String[] hostNameList = new String[numOfServers];
    for(int index = 0; index < numOfServers; ++index) {
      hostNameList[index] = startClusterManager(index);
    }
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      return null;
    }
    
    return hostNameList;
  } 
  
  protected String startClusterManager(int serverNum) throws IOException {
    PleiadesConfiguration thisConf = new PleiadesConfiguration(conf);  
    String hostName;
    try {
      thisConf.set("clusterManager.port", 8000 + serverNum);
      thisConf.set("clusterManager.handler.count", 3);
      thisConf.set("clustermanager.heartbeat.port", 11000 + serverNum);
      ClusterManager clusterManager = new ClusterManager("conf/session.prop", thisConf);
      hostName = clusterManager.getHostName();
      
      clusterManagers.put(hostName, clusterManager);
      clusterManagerConfs.put(hostName, thisConf);
    } catch (Exception e) {
      e.printStackTrace(System.out);
      throw new IOException(e.getMessage());
    }
    
    Thread t = new Thread(clusterManager, "ClusterManager-" + this.clusterManagers.size());
    t.start();
    this.clusterManagerThreads.put(hostName, t);
    
    return hostName;
  }
  
  protected void stopClusterManagers() throws IOException {
    LOG.debug("Cluster Manager shutdown");    
    for(Map.Entry<String, ClusterManager> entry: clusterManagers.entrySet()) {
      entry.getValue().shutdown();
    }    
    for(Map.Entry<String, Thread> entry: clusterManagerThreads.entrySet()) {
      entry.getValue().interrupt();
    }
    try {
      Thread.sleep(2 * 1000);
    } catch (InterruptedException e) {
    }
  }
  
  protected void startEventManagers(int numOfServers) throws IOException {
    for(int index = 0; index < numOfServers; ++index) {
      startEventManager(index);
    }
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      return;
    }
  }
  
  protected void startEventManager(int serverNum) throws IOException {
    try {
      conf.set("eventManager.port", 10000 + serverNum);
      conf.set("eventManager.handler.count", 5);
      eventManager = new EventManager(conf);
      this.eventManagers.put(eventManager.getHostName(), eventManager);
    } catch (Exception e) {
      e.printStackTrace(System.out);
      throw new IOException(e.getMessage());
    }
    Thread t = new Thread(eventManager, "EventManager-" + this.eventManagers.size());
    t.start();
    this.eventManagerThreads.put(eventManager.getHostName(), t);    
  }
  
  protected void stopEventManagers() throws IOException {
    LOG.debug("Event Managers are shutdown");    
    for(Map.Entry<String, EventManager> entry: eventManagers.entrySet()) {
      entry.getValue().shutdown();
    }    
    for(Map.Entry<String, Thread> entry: eventManagerThreads.entrySet()) {
      entry.getValue().interrupt();
    }
    try {
      Thread.sleep(2 * 1000);
    } catch (InterruptedException e) {
    }
  }
  
  public void startFirstRouter() throws Exception {
    firstRouter = new GossipRouter(portOfRouter1, bind_addr);
    firstRouter.start();
    
  }

  public void stopFirstRouter() {
    firstRouter.stop();
  }  
  
  public void dumpFirstRouter() {
    if(firstRouter != null) {
      LOG.info(firstRouter.dumpRoutingTable());  
    } else {
      LOG.info("First Router is not initialized");
    }
  }
  
  public void dumpSecondRouter() {
    if(secondRouter != null) {
      LOG.info(secondRouter.dumpRoutingTable());  
    } else {
      LOG.info("Second Router is not initialized");
    }
  }
  
  protected void setUp() throws Exception {
    super.setUp(); 
    startCluster();
    startClient();     
  }

  public void tearDown() throws Exception {
    try {
      super.tearDown();
      stopClient();
      stopCluster();
      Thread.sleep(1000);
    } finally {      
      Thread.sleep(2000);
    }
  }
  
  public static LockMetaData[] createMetaData(String path, int count) {    
    LockMetaData[] metaDataList = new LockMetaData[count];
    for(int i=0; i < count; ++i) {
      metaDataList[i] = new LockMetaData(path + i);
      metaDataList[i].setContentKey("contentKey:" + i + "_" + System.currentTimeMillis());
    }    
    return metaDataList;
  }
  
  public static boolean getContentsResult(String contents) {
    LOG.info(contents);
    boolean isSuccess = false;
    if(!contents.equals(""))
      isSuccess = true;
    return isSuccess;
  }
  
  public static void printMetaData(LockMetaData metaData, boolean printContents) {
    LOG.info("--------------Start Printing MetaData---------------");
    LOG.info(metaData.getCreator());
    LOG.info(metaData.getContentKey());
    LOG.info(metaData.getGenerationNumber());
    LOG.info("--------------End Printing MetaData----------------");
  }
}
